package com.fyf.game.net.server;

import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicLong;
import org.je.log.JLog;
import com.fyf.game.net.DataAnalyzer;
import com.fyf.game.net.NioUtil;
import com.fyf.game.net.ReceiveHandler;
import com.fyf.game.net.SelectManager;

/**
 * @author fanyifeng 2013-1-8
 */
public class Transporter extends Thread
{
	private SelectManager	selMgr;
	private boolean			closed;
	private ByteBuffer		readTemp;
	private ReceiveHandler	handler;
	private RunEnvir		env;

	public final AtomicLong	registerChannelCount	= new AtomicLong(0);

	public Transporter(ReceiveHandler receiveHandler) {
		handler = receiveHandler;
		closed = false;
		selMgr = new SelectManager();
		readTemp = ByteBuffer.allocate(1024);
	}

	Transporter setEnvir(RunEnvir env) {
		this.env = env;
		return this;
	}

	public void run() //
	{
		while (!closed) //
		{
			selMgr.setNeedWakeup();
			synchronized(this) {}

			if (selMgr.select() <= 0) {
				continue;
			}

			Iterator<SelectionKey> it = selMgr.keys();

			while (it.hasNext()) {
				SelectionKey key = it.next();
				it.remove();
				if (!key.isValid()) {
					NioUtil.cancelAndClose(key);
					continue;
				}

				try {
					if (key.isReadable()) {
						doRead(key);
					} else if (key.isWritable()) {
						doWrite(key);
					}
				} catch (Exception ex) {
					NioUtil.cancelAndClose(key);
					JLog.error(ex);
					ex.printStackTrace();
				}
			}
		}
		selMgr.close();
		env.closeCount.incrementAndGet();// 退出计数
	}

	public void doRead(SelectionKey key) {
		SocketChannel channel = (SocketChannel) key.channel();
		DataManager dm = (DataManager) key.attachment();
		readTemp.clear();
		int readLen = 0;

		try {
			readLen = channel.read(readTemp);
		} catch (Exception ex) {
			env.errorReadCount.incrementAndGet();
			NioUtil.cancelAndClose(key);
			env.senderBox.remove(dm.senderID());
			JLog.error(ex);
			return;
		}

		if (readLen < 0) {
			env.errorReadCount.incrementAndGet();
			NioUtil.cancelAndClose(key);
			env.senderBox.remove(dm.senderID());
			return;
		}

		DataAnalyzer ana = dm.analyzer();
		ana.saveAndAnalyse(readTemp);
		if (ana.needClose()) {// 正常退出
			dm.put("");
			key.interestOps(SelectionKey.OP_WRITE);
		}
	}

	public void doWrite(SelectionKey key) {
		SocketChannel channel = (SocketChannel) key.channel();
		DataManager dm = (DataManager) key.attachment();
		ByteBuffer data = dm.getWriteData();
		if (data == null) {
			return;
		}
		try {
			channel.write(data);
		} catch (Exception ex) {
			env.errorWriteCount.incrementAndGet();
			NioUtil.cancelAndClose(key);
			env.senderBox.remove(dm.senderID());
			JLog.error(ex);
			return;
		}

		if (data.hasRemaining()) {
			return;
		}

		if (data.capacity() == 10) {// 表明是退出语句
			env.successOutLine.incrementAndGet();
			NioUtil.cancelAndClose(key);
			env.senderBox.remove(dm.senderID());
		} else {
			dm.removeFirstWriteData(true);
		}
	}

	public void addChannel(SocketChannel ch) {
		DataManager dm = new DataManager(selMgr);
		DataAnalyzer ana = new DataAnalyzer(handler, new MsgSender(dm, env.senderBox), env.successReadMsgCount);
		registerChannelCount.incrementAndGet();
		try {
			synchronized(this) {
				selMgr.wakeup();
				SelectionKey key = selMgr.register(ch, SelectionKey.OP_READ);
				dm.setAnalyzerAndKey(ana, key);
				key.attach(dm);
			}
		} catch (Exception ex) {
			NioUtil.close(ch);
			JLog.error(ex);
		}
	}

	public void close() {
		closed = true;
		selMgr.wakeup();
	}
}
